#pragma once

#include <unordered_set>
#include "../comm/message.hpp"
#include "../comm/net.hpp"

namespace Server
{
    using namespace MuduoNet;
    class TopicManger
    {
    public:
        using ptr = std::shared_ptr<TopicManger>;
        TopicManger(){}
        void OnTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            TopicOptype topicop = msg->GetOptype();
            bool ret = true;
            switch (topicop)
            {
                // 主题创建
            case TopicOptype::TOPIC_CREATE:
                TopicCreate(conn, msg);
                break;
                // 主题删除
            case TopicOptype::TOPIC_REMOVE:
                TopicRemove(conn, msg);
                break;
            // 主题注册
            case TopicOptype::TOPIC_SUBSCRIBE:
                ret = TopicSubscribe(conn, msg);
                break;
                // 主题取消订阅
            case TopicOptype::TOPIC_CANCEL:
                TopicCancel(conn, msg);
                break;
                // 主题消息发布
            case TopicOptype::TOPIC_PUBLISH:
                ret = TopicPublish(conn, msg);
                break;
            default:
                return ErrorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE);
            }
            if (!ret) return ErrorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
            printf("ret = %d\n", ret);
            return TopicResponse(conn, msg);
        }

        void OnShutDown(const BaseConnection::ptr conn)
        {
            // 消息发布者断开连接后，不需要进行任何操作，消息订阅者断开连接，需要删除刮玻璃数据
            ////1.判断断开连接是否是订阅者，不是的直接返回
            vector<Topic::ptr> topics;
            Subscriber::ptr subscriber;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _subscribers.find(conn);
                if (it == _subscribers.end())
                {
                    return;
                }
                subscriber = it->second;
                // 2.获取到订阅者退出
                for (auto &topic_name : subscriber->topics)
                {
                    auto topic_it = _topics.find(topic_name);
                    if (topic_it == _topics.end())
                        continue;
                    topics.push_back(topic_it->second);
                }
                // 从订阅者映射信息来看，删除订阅者
                _subscribers.erase(it);
            }
            // 从受影响的主题中，移除订阅者
            for (auto &e : topics)
            {
                e->RemoveSubscriber(subscriber);
            }
        }

    private:
        void ErrorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, RCode rcode)
        {
            auto msg_rsp = MessageFactory::create<Message::TopicResponse>();
            msg_rsp->SetId(msg->GetRid());
            msg_rsp->SetMtype(Mtype::RSP_TOPIC);
            msg_rsp->SetRcode(rcode);
            conn->Send(msg_rsp);
        }
        void TopicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            auto msg_rsp = MessageFactory::create<Message::TopicResponse>();
            msg_rsp->SetId(msg->GetRid());
            msg_rsp->SetMtype(Mtype::RSP_TOPIC);
            msg_rsp->SetRcode(RCode::RCODE_OK);
            conn->Send(msg_rsp);
        }
        void TopicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            // 创建一个主题对象，添加映射管理
            std::unique_lock<mutex> lock(_mutex);
            std::string topic_name = msg->GetTopicKey();
            auto topic = std::make_shared<Topic>(topic_name);
            _topics.insert(make_pair(topic_name, topic));
        }

        void TopicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            // 1.查看当前主题，有哪些订阅者，然后从订阅者中将主题信息删除掉

            std::string topic_name = msg->GetTopicKey();
            std::unordered_set<Subscriber::ptr> subscribers;
            {
                std::unique_lock<mutex> lock(_mutex);
                // 删除主题之前，先找出会受影响的订阅者
                auto it = _topics.find(topic_name);
                if (it == _topics.end())
                    return;
                subscribers = it->second->subscribers;
                _topics.erase(it); // 删除当前主题映射关系
            }
            // 2.删除主题信息数据，--主题名称与主题对象的映射关系
            for (auto subsciber : subscribers)
            {
                subsciber->RemoveTopic(topic_name);
            }
        }
        bool TopicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            // 先找到主题对象，以及订阅者对象
            // 没有对象就报错；没有就要创建一个订阅者
            Topic::ptr topic;
            Subscriber::ptr subscriber;
            {
                std::unique_lock<mutex> lock(_mutex);
                auto topic_it = _topics.find(msg->GetTopicKey());
                if (topic_it == _topics.end())
                {
                    return false;
                }
                topic = topic_it->second;
                auto sub_it = _subscribers.find(conn);
                if (sub_it != _subscribers.end())
                {
                    subscriber = sub_it->second;
                }
                else
                {
                    // 没有连接就要创建。
                    subscriber = make_shared<Subscriber>(conn);
                    _subscribers.insert(make_pair(conn, subscriber));
                }
            }
            // 在主题中，新增一个订阅者对象关联的连接，订阅者对象新增一个主题
            topic->AppendSubscriber(subscriber);
            subscriber->AppendTopic(msg->GetTopicKey());
            return true;
        }
        void TopicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            // 1. 先找出主题对象，和订阅者对象
            // 2. 从主题对象中删除当前的订阅者连接；   从订阅者信息中删除所订阅的主题名称
            Topic::ptr topic;
            Subscriber::ptr subscriber;
            {
                std::unique_lock<mutex> lock(_mutex);
                auto topic_it = _topics.find(msg->GetTopicKey());
                if (topic_it != _topics.end())
                {
                    topic = topic_it->second;
                }
                auto sub_it = _subscribers.find(conn);
                if (sub_it != _subscribers.end())
                {
                    subscriber = sub_it->second;
                }
            }
            if (subscriber)
                subscriber->RemoveTopic(msg->GetTopicKey());
            if (topic && subscriber)
                topic->RemoveSubscriber(subscriber);
        }
        bool TopicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            Topic::ptr topic;

            {
                std::unique_lock<mutex> lock(_mutex);
                auto topic_it = _topics.find(msg->GetTopicKey());
                if (topic_it == _topics.end())
                {
                    return false;
                }
                topic = topic_it->second;
            }
            topic->PushMessage(msg);
            return true;
        }

    private:
        struct Subscriber
        {
            using ptr = std::shared_ptr<Subscriber>;
            std::mutex _mutex;
            BaseConnection::ptr conn;
            std::unordered_set<std::string> topics;

            Subscriber(const BaseConnection::ptr &c) : conn(c) {}
            // 订阅主题时调用
            void AppendTopic(const std::string &topic_name)
            {
                std::unique_lock<mutex> lock(_mutex);
                topics.insert(topic_name);
            }
            // 主题被删除，或者取消订阅的时候调用
            void RemoveTopic(const std::string &topic_name)
            {
                std::unique_lock<mutex> lock(_mutex);
                topics.erase(topic_name);
            }
        };
        struct Topic
        {
            using ptr = std::shared_ptr<Topic>;
            std::mutex _mutex;
            std::string topic_name;
            std::unordered_set<Subscriber::ptr> subscribers;

            Topic(const std::string &name) : topic_name(name) {}
            // 新增订阅
            void AppendSubscriber(const Subscriber::ptr &subscriber)
            {
                std::unique_lock<mutex> lock(_mutex);
                subscribers.insert(subscriber);
            }
            // 取消订阅
            void RemoveSubscriber(const Subscriber::ptr &subscriber)
            {
                std::unique_lock<mutex> lock(_mutex);
                subscribers.erase(subscriber);
            }
            // 收到消息发布请求调用
            void PushMessage(const BaseMessage::ptr &msg)
            {
                std::unique_lock<mutex> lock(_mutex);
                for (auto subscriber : subscribers)
                {
                    subscriber->conn->Send(msg);
                }
            }
        };

    private:
        std::mutex _mutex;
        unordered_map<std::string, Topic::ptr> _topics;
        unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;
    };
}